package com.zhang.table;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: 读取文件系统数据
 * @author: zhang
 * @date: 2022/2/10 19:43
 */
public class FileSystemSource {
    public static void main(String[] args) throws Exception {
        // TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // TODO 2 使用connect 方式读取文本数据
        tableEnvironment.connect(new FileSystem().path("input/sensor.txt"))
                .withSchema(new Schema()
                        .field("id", DataTypes.STRING())
                        .field("temp", DataTypes.INT()))
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                .createTemporaryTable("sensor");

        //TODO 3 将连接器应用，转换为表
        Table sensor = tableEnvironment.from("sensor");

        //TODO 4 查询
        Table select = sensor.groupBy($("id"))
                .select($("id"), $("id").count().as("cnt"));

        // TODO 5 转换流输出
        tableEnvironment.toRetractStream(select, Row.class).print();


        env.execute();
    }
}
